// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/base/feed/ws/ws_connection.h"

#include <regex>

#include <boost/algorithm/string.hpp>
#include <boost/asio/ip/address.hpp>
#include <boost/beast/core/buffers_to_string.hpp>
#include <boost/beast/core/multi_buffer.hpp>
#include <boost/beast/websocket/rfc6455.hpp>  // websocket::close_code
#include <boost/beast/websocket/stream.hpp>
#include <boost/beast/http/message.hpp>
#include <boost/date_time/gregorian/gregorian_types.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>
#include <nlohmann/json.hpp>

#include "coin2/exchange/base/api_util/raw_subscriber_util.h"
#include "coin2/exchange/base/api_util/zlib_decompress.h"
#include "coin2/exchange/base/executor/live_executor.h"
#include "coin2/exchange/base/executor/topic_distributor.h"
#include "coin2/exchange/base/feed/ws/feed_subscription_spec.h"
#include "coin2/exchange/base/feed/ws/request_generator.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/feed.pb.h"

namespace coin2::exchange::base::feed {
namespace impl {

WsConnection::WsConnection(
    const WsConnectionConfig& config,
    std::function<void()> fail_callback,
    executor::LiveAsyncExecutor* exec,
    RequestGenerator* request_generator,
    std::shared_ptr<LatencyExtractor> latency_extractor,
    bool enable_telemetry)
    : config_(config),
      fail_callback_(std::move(fail_callback)),
      running_(false),
      ws_closed_(false),
      exec_(exec),
      strand_(*exec->RawIoc()),
      ctx_(boost::asio::ssl::context::sslv23_client),
      resolver_(strand_),
      use_ssl_(!config.url.disable_ssl),
      scheduler_(new Scheduler(&strand_)),
      request_generator_(request_generator),
      latency_extractor_(latency_extractor),
      enable_telemetry_(enable_telemetry) {
  if (config.mea_config->unauthorized_limit.subscription) {
    unauthorized_subscription_limit_ = std::make_shared<AsyncLeakyBucket>(
        &strand_,
        config.mea_config->unauthorized_limit.subscription->first,
        config.mea_config->unauthorized_limit.subscription->second);
  }

  if (config.mea_config->unauthorized_limit.command) {
    unauthorized_command_limit_ = std::make_shared<AsyncLeakyBucket>(
        &strand_,
        config.mea_config->unauthorized_limit.command->first,
        config.mea_config->unauthorized_limit.command->second);
  }

  if (use_ssl_) {
    ws_ssl_.reset(new WebsocketSsl(strand_, ctx_));
  } else {
    ws_nossl_.reset(new WebsocketNoSsl(strand_));
  }
}

WsConnection::~WsConnection() {
  running_ = false;
  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", destroy";
}

void WsConnection::RunAsync() {
  running_ = true;
  std::weak_ptr<WsConnection> self(shared_from_this());
  boost::asio::spawn(strand_, [self](boost::asio::yield_context yc) {
    if (auto s = self.lock()) {
      s->Run(yc);  // NOLINT
    } else {
      LOG(ERROR) << "WsConnection::RunAsync : object is destroyed";
    }
  });
}

std::string WsConnection::Decompress(const std::string& s) const {
  std::string outstring;
  bool ok = false;
  if (config_.mea_config->decompress == "deflate") {
    ok = deflate_string(s, &outstring);
  } else if (config_.mea_config->decompress == "gzip") {
    ok = decompress_gzip_string(s, &outstring);
  } else {
    return s;
  }

  if (ok) {
    return outstring;
  } else {
    LOG_EVERY_N(ERROR, 1000) << config_.topic << "_" << config_.url.name
        << " use original data as decompress error, " << outstring;
    return s;
  }
}

std::optional<std::string> WsConnection::HandlePing(const std::string& s) const {
  if (config_.mea_config->handle_ping_msg) {
    if (s.rfind(R"({"ping":)") == 0 ||
        s.rfind(R"({"op":"ping")") == 0 ||
        s.find(R"("primus::ping::)") == 0) {
      std::optional<std::string> resp;
      if (config_.mea_config->mea.exchange == coin::proto::ExchangeType::Lbank) {
        resp = boost::replace_all_copy(s, "ping", "pong");
      } else {
        resp = get_pong(s);
      }
      return resp;
    } else if (config_.mea_config->mea.exchange == coin::proto::ExchangeType::Crypto &&
               s.size() < 100) {
      // crypto heartbeat size is small
      if (s.find(R"("method": "public/heartbeat")")) {
        auto resp = boost::replace_all_copy(s, "heartbeat", "respond-heartbeat");
        return resp;
      }
    }

    return {};
  }
  return {};
}

std::string WsConnection::GetSubscriptionRequest(const std::string& symbol_channel) {
  std::string req = config_.mea_config->subscription_req;
  if (req.find("{json_list}") != std::string::npos) {
    nlohmann::json j;
    j.push_back(symbol_channel);
    boost::replace_all(req, "{json_list}", j.dump());
  } else {
    boost::replace_all(req, "{channel}", symbol_channel);
  }
  return req;
}

std::vector<std::string> WsConnection::GetSubscriptionRequests() const {
  return request_generator_->GenerateRequests(
      config_.url.channels,
      config_.symbols,
      config_.mea_config->channels,
      config_.mea_config->subscription_req);
}

template <typename T>
void WsConnection::CheckAndSetDecorator(T& stream) {
  if (config_.require_header) {
    stream.set_option(
        boost::beast::websocket::stream_base::decorator(
          std::bind(&WsConnection::InsertHeader, this, std::placeholders::_1)));
  }
  if (config_.mea_config->permessage_deflate) {
    boost::beast::websocket::permessage_deflate opt;
    opt.client_enable = true;
    stream.set_option(opt);
  }
  stream.set_option(boost::beast::websocket::stream_base::decorator(
      [](boost::beast::http::request<boost::beast::http::empty_body>& req) {
        LOG(INFO) << "Customized ws upgrade request: " << req;
      }));
}

void WsConnection::InsertHeader(boost::beast::websocket::request_type& req) {
  CHECK(!config_.key_filepath.empty());
  auto headers = request_generator_->GenerateHeader(config_.key_filepath);

  for (auto& header : headers) {
    req.set(header.first, header.second);
  }
}

std::vector<std::string> WsConnection::GetUnsubscriptionRequests() const {
  return request_generator_->GenerateRequests(
      config_.url.channels,
      config_.symbols,
      config_.mea_config->channels,
      config_.mea_config->subscription_req);
}

std::vector<std::string> WsConnection::GetUnsubscriptionRequest(
    const std::vector<std::string> symbols) const {
  return request_generator_->GenerateRequests(
      config_.url.channels,
      symbols,
      config_.mea_config->channels,
      config_.mea_config->unsubscription_req);
}

void WsConnection::Unsubscribe(
    const std::vector<std::string>& symbols) {
  // remove symbol from configuration for periodic resubscribe
  for (auto iter = config_.symbols.begin(); iter != config_.symbols.end(); ) {
    auto pos = std::find(symbols.begin(), symbols.end(), *iter);
    if (pos != symbols.end()) {
      iter = config_.symbols.erase(iter);
    } else {
      ++iter;
    }
  }

  // cache unsubscribed symbol for unsubscribe response
  removed_symbols_.insert(removed_symbols_.end(), symbols.begin(), symbols.end());
  
  const auto requests = GetUnsubscriptionRequest(symbols);

  std::weak_ptr<WsConnection> self(shared_from_this());
  boost::asio::spawn(
      strand_, [self, requests](boost::asio::yield_context yc) {
        if (auto s = self.lock()) {
          for (const auto& req : requests) {
            s->Subscribe(yc, req);
          }
        } else {
          LOG(ERROR) << "WsConnection::UnsubscribeAsync : object is destroyed";
        }
      });
}

void WsConnection::PostUnsubcribe(const std::vector<std::string>& symbols) {
  std::weak_ptr<WsConnection> self(shared_from_this());
  boost::asio::post(
      strand_,
      [self, symbols] {
        if (auto s = self.lock()) {
          s->Unsubscribe(symbols);
        }
      });
}

void WsConnection::SubscribeAll(boost::asio::yield_context yield) {
  if (config_.require_auth) {
    Send(request_generator_->GenerateAuthRequest(config_.key_filepath));
  }
  if (unauthorized_subscription_limit_) {
    boost::beast::error_code ec;
    unauthorized_subscription_limit_->PostCoro(yield[ec]);
  }

  const auto requests = GetSubscriptionRequests();
  for (const auto& req : requests) {
    Subscribe(yield, req);
  }
}

void WsConnection::Subscribe(boost::asio::yield_context yield, const std::string& request) {
  boost::beast::error_code ec;

  nlohmann::json msg = {
      {"compartment", "feed"},
      {"component", "WsConnection."s + config_.topic + "_" + config_.url.name},
      {"mea", config_.mea_config->mea.String()},
      {"request", request},
  };
  LOG(INFO) << msg.dump();

  if (unauthorized_subscription_limit_) {
    unauthorized_subscription_limit_->PostCoro(yield[ec]);
  }
  // sub op consumes command
  if (unauthorized_command_limit_) {
    unauthorized_command_limit_->PostCoro(yield[ec]);
  }
  Send(request);
}

void WsConnection::UnsubscribeAll(boost::asio::yield_context yield) {
  boost::beast::error_code ec;
  const auto requests = GetUnsubscriptionRequests();

  nlohmann::json msg = {
      {"compartment", "feed"},
      {"component", "WsConnection."s + config_.topic + "_" + config_.url.name},
      {"mea", config_.mea_config->mea.String()},
      {"requests", requests},
  };
  LOG(INFO) << msg.dump();

  for (const auto& req : requests) {
    // unsub op consumes command
    if (unauthorized_command_limit_) {
      unauthorized_command_limit_->PostCoro(yield[ec]);
    }
    Send(req);
  }
}

bool WsConnection::IsUnsubscriptionResponse(const std::string& msg, std::string* symbol_channel)
    const {
  if (config_.mea_config->unsubscription_resp.empty()) {
    return false;
  }

  std::smatch match;
  try {
    std::regex re(config_.mea_config->unsubscription_resp);
    if (std::regex_search(msg, match, re) && match.size() > 1) {
      auto matched = match.str(1);
      if (IsSymbolRemoved(matched)) {
        return false;
      }

      *symbol_channel = matched;
      return true;
    }
  } catch (const std::regex_error& e) {
    LOG(ERROR) << e.what();
  }
  return false;
}

bool WsConnection::IsSymbolRemoved(const std::string& symbol_channel) const {
  for (auto& symbol : removed_symbols_) {
    if (symbol_channel.find(symbol) != std::string::npos) {
      return true;
    }
  }

  return false;
}

int64_t WsConnection::Delay() {
  using boost::gregorian::date_duration;
  using boost::posix_time::ptime;
  using boost::posix_time::second_clock;

  ptime now(second_clock::universal_time());
  ptime midnight(now.date() + date_duration(1));
  return (midnight - now).total_nanoseconds();
}

void WsConnection::Run(boost::asio::yield_context yield) {
  Run(yield, config_.url.host, config_.url.port, config_.url.path);
}

void WsConnection::Run(boost::asio::yield_context yield,
                       const std::string& host, const std::string& port,
                       const std::string& path) {
  DLOG(INFO) << config_.topic << "_" << config_.url.name
             << ", running: " << running_;
  boost::beast::error_code ec;

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", resolve";
  // Look up the domain name
  auto const results = resolver_.async_resolve(host, port, yield[ec]);
  if (ec) {
    return fail(
        &yield, ec,
        std::string("cannot resolve ip for host=") + host + " port=" + port);
  }

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", connect";
  boost::asio::ip::tcp::socket* socket = nullptr;
  if (use_ssl_) {
    socket = &boost::beast::get_lowest_layer<WebsocketSsl>(*ws_ssl_).socket();
  } else {
    socket = &boost::beast::get_lowest_layer<WebsocketNoSsl>(*ws_nossl_).socket();
  }
  if (!config_.url.bind_ip.empty()) {
    std::string ip = config_.url.bind_ip;
    LOG(INFO) << "Feed worker " << config_.worker_id << " bind to " << ip;
    socket->open(boost::asio::ip::tcp::v4());
    socket->bind(
        boost::asio::ip::tcp::endpoint(boost::asio::ip::make_address(ip), 0));

    DLOG(INFO) << "Before connecting socket has local endpoint: "
               << socket->local_endpoint().address().to_string() << ":"
               << socket->local_endpoint().port();

    std::weak_ptr<WsConnection> self(shared_from_this());

    // To use a specific IP, we have to open the socket first and bind.
    // But if calling boost::asio::async_connect() it will close the socket
    // and restart.
    // So the solution is to use socket.async_connect() instead.
    // https://stackoverflow.com/questions/20890703/how-to-use-a-specfic-networkinterface-ip-with-boost-asio-sockets/20975511
    socket->async_connect(
        *results.begin(),
        [self, socket, host, path](const boost::beast::error_code& ec) {
          if (auto s = self.lock()) {
            if (ec) {
              return s->fail_async1(ec, std::string("cannot connect ip"));
            }
            DLOG(INFO) << "After connecting socket has local endpoint: "
                       << socket->local_endpoint().address().to_string() << ":"
                       << socket->local_endpoint().port();

            boost::asio::spawn(
                s->strand_, [self, host, path](boost::asio::yield_context yc) {
                  if (auto s = self.lock()) {
                    s->RunAfterConnect(yc, host, path);  // NOLINT
                  } else {
                    LOG(ERROR) << "WsConnection object is already destroyed";
                  }
                });
          } else {
            LOG(ERROR) << "WsConnection object is already destroyed";
          }
        });
  } else {
    // Make the connection on the IP address we get from a lookup
    if (use_ssl_) {
      boost::beast::get_lowest_layer(*ws_ssl_).async_connect(results,
                                                             yield[ec]);
    } else {
      boost::beast::get_lowest_layer(*ws_nossl_)
          .async_connect(results, yield[ec]);
    }
    if (ec) {
      return fail(&yield, ec, std::string("cannot connect ip"));
    }
    RunAfterConnect(yield, host, path);
  }
}

void WsConnection::RunAfterConnect(boost::asio::yield_context yield,
                                   const std::string& host,
                                   const std::string& url_path) {
  boost::beast::error_code ec;
  // auto host = config_.url.host;
  auto path = url_path;
  if (config_.mea_config->require_regenerate_path) {
    path = request_generator_->RegeneratePath(url_path, config_.key_filepath);
  }

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", set_option";
  if (use_ssl_) {
    ws_ssl_->set_option(boost::beast::websocket::stream_base::timeout{
        std::chrono::seconds(5),  // handshake timeout
        std::chrono::nanoseconds(
            config_.ws_config->timeout()),  // idle timeout
        // if true, we wait for another timeout. otherwise, immediately
        // disconnects.
        config_.url.support_ping_frame,
    });
    CheckAndSetDecorator(*ws_ssl_);
  } else {
    ws_nossl_->set_option(boost::beast::websocket::stream_base::timeout{
        std::chrono::seconds(5),  // handshake timeout
        std::chrono::nanoseconds(
            config_.ws_config->timeout()),  // idle timeout
        // if true, we wait for another timeout. otherwise, immediately
        // disconnects.
        config_.url.support_ping_frame,
    });
    CheckAndSetDecorator(*ws_nossl_);
  }

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", handshake";
  boost::beast::http::response<boost::beast::http::string_body> resp;
  if (use_ssl_) {
    // https://stackoverflow.com/questions/59224873/boost-beast-handshake-sslv3-alert-handshake-failure-error
    if (config_.mea_config->set_tlsext_host_name) {
      if (!SSL_set_tlsext_host_name(ws_ssl_->next_layer().native_handle(),
                                    host.c_str())) {
        boost::system::error_code ec{static_cast<int>(::ERR_get_error()),
                                     boost::asio::error::get_ssl_category()};
        return fail(&yield, ec, "ssl_handshake_hostname");
      }
    }

    DLOG(INFO) << config_.topic << "_" << config_.url.name
               << ", ssl_handshake call";
    ws_ssl_->next_layer().async_handshake(boost::asio::ssl::stream_base::client,
                                          yield[ec]);
    DLOG(INFO) << config_.topic << "_" << config_.url.name
               << ", ssl_handshake done";
    if (ec) {
      return fail(&yield, ec, "ssl_handshake");
    }
    ws_ssl_->async_handshake(resp, host, path, yield[ec]);
  } else {
    ws_nossl_->async_handshake(resp, host, path, yield[ec]);
  }
  LOG(INFO) << "ws handshake response: " << resp << std::endl;
  if (ec) {
    return fail(&yield, ec, "handshake");
  }

  // Schedule periodic task that's executed at the beginning of each day.
  // This task will then be executed periodically.
  if (config_.mea_config->use_periodic_resubscribe) {
    DLOG(INFO) << config_.topic << "_" << config_.url.name << ", resubscribe";
    scheduler_->Start();
    std::weak_ptr<WsConnection> self(shared_from_this());
    scheduler_->AddPeriodicTask(
        [self] {
          if (auto s = self.lock()) {
            s->Unsubscribe();
          } else {
            LOG(ERROR) << "WsConnection handler is already destroyed";
          }
        },
        config_.mea_config->resubscribe_interval_ns, "unsubscribe", Delay());
  }

  {
    DLOG(INFO) << config_.topic << "_" << config_.url.name << ", subscribe";
    std::weak_ptr<WsConnection> self(shared_from_this());
    boost::asio::spawn(strand_, [self](boost::asio::yield_context yc) {
      if (auto s = self.lock()) {
        s->SubscribeAll(std::move(yc));
      } else {
        LOG(ERROR) << "WsConnection::RunAsync : object is destroyed";
      }
    });
  }

  SchedulePingPong();

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", read";

  boost::beast::multi_buffer buffer;

  while (running_) {
    boost::asio::ip::tcp::socket* socket = nullptr;
    if (use_ssl_) {
      socket = &boost::beast::get_lowest_layer<WebsocketSsl>(*ws_ssl_).socket();
    } else {
      socket =
          &boost::beast::get_lowest_layer<WebsocketNoSsl>(*ws_nossl_).socket();
    }
    int64_t pre_read_available = 0;
    int64_t pre_read_timestamp = 0;
    if (enable_telemetry_) {
      pre_read_available = socket->available(ec);
      if (ec) {
        return fail(&yield, ec, "fail preread available ws");
      }
      pre_read_timestamp = executor::TopicRecordProcessor::GetTimestamp();
    }
    std::size_t bytes_transferred;
    if (use_ssl_) {
      bytes_transferred = ws_ssl_->async_read(buffer, yield[ec]);
    } else {
      bytes_transferred = ws_nossl_->async_read(buffer, yield[ec]);
    }
    if (ec) {
      return fail(&yield, ec, "fail reading ws");
    }
    int64_t post_read_available = 0;
    if (enable_telemetry_) {
      post_read_available = socket->available(ec);
      if (ec) {
        return fail(&yield, ec, "fail prostread available ws");
      }
    }
    boost::ignore_unused(bytes_transferred);
    num_messages_++;

    std::string s = Decompress(boost::beast::buffers_to_string(buffer.data()));
    buffer.consume(buffer.size());

    std::string symbol_channel;
    if (const auto pong = HandlePing(s)) {
      Send(*pong);
    } else if (IsUnsubscriptionResponse(s, &symbol_channel)) {
      const auto request = GetSubscriptionRequest(symbol_channel);

      std::weak_ptr<WsConnection> self(shared_from_this());
      boost::asio::spawn(
          strand_, [self, request](boost::asio::yield_context yc) {
            if (auto s = self.lock()) {
              s->Subscribe(std::move(yc), request);
            } else {
              LOG(ERROR) << "WsConnection::RunAsync : object is destroyed";
            }
          });
    } else {
      auto packet = std::shared_ptr<TopicRecordPacket>(new TopicRecordPacket{
          config_.topic, 0, executor::TopicRecordProcessor::GetTimestamp(), 0,
          0, 0, pre_read_available, pre_read_timestamp, post_read_available,
          s});
      exec_->injector().push(packet);
      if (latency_extractor_) {
        latency_extractor_->PostExtractLatency(packet);
      }
    }
  }

  DLOG(INFO) << config_.topic << "_" << config_.url.name << ", close";
  // Close the WebSocket connection
  if (!ws_closed_) {
    ws_closed_ = true;
    if (use_ssl_) {
      ws_ssl_->async_close(boost::beast::websocket::close_code::normal,
                           yield[ec]);
    } else {
      ws_nossl_->async_close(boost::beast::websocket::close_code::normal,
                             yield[ec]);
    }
    if (ec) {
      return fail(&yield, ec, "gracious async_close failed");
    }
  }
}

void WsConnection::fail(
    boost::asio::yield_context* yield,
    boost::system::error_code ec,
    const std::string& what) {
  if (!running_) {
    return;
  }
  running_ = false;
  scheduler_->Stop();

  nlohmann::json msg = {
      {"compartment", "feed"},
      {"component", "WsConnection."s + config_.topic + "_" + config_.url.name},
      {"mea", config_.mea_config->mea.String()},
      {"presto_error", what},
      {"beast_error", ec.message()},
      {"num_feed_messages", num_messages_},
      {"num_subscription_sent", num_subscription_sent_},
  };
  LOG(ERROR) << msg.dump();

  // until async_close is done, the coroutine is alive
  if (!ws_closed_) {
    ws_closed_ = true;
    if (use_ssl_) {
      ws_ssl_->async_close(
          boost::beast::websocket::close_code::normal,
          (*yield)[ec]);  // ignore error
    } else {
      ws_nossl_->async_close(
          boost::beast::websocket::close_code::normal,
          (*yield)[ec]);  // ignore error
    }
  }

  // we are holding a shared_ptr
  fail_callback_();
}

void WsConnection::fail_async1(boost::system::error_code ec, const std::string& what) {
  if (!running_) {
    return;
  }
  running_ = false;
  scheduler_->Stop();

  nlohmann::json msg = {
      {"compartment", "feed"},
      {"component", "WsConnection."s + config_.topic + "_" + config_.url.name},
      {"mea", config_.mea_config->mea.String()},
      {"presto_error", what},
      {"beast_error", ec.message()},
      {"num_feed_messages", num_messages_},
      {"num_subscription_sent", num_subscription_sent_},
  };
  LOG(ERROR) << msg.dump();

  // until async_close is done, the coroutine is alive
  if (!ws_closed_) {
    ws_closed_ = true;
    std::weak_ptr<WsConnection> self(shared_from_this());
    if (use_ssl_) {
      ws_ssl_->async_close(
          boost::beast::websocket::close_code::normal,
          [self](boost::system::error_code ec) {
            if (auto s = self.lock()) {
              s->fail_async2(ec);
            } else {
              LOG(ERROR) << "WsConnection is already destroyed";
            }
          });
    } else {
      ws_nossl_->async_close(
          boost::beast::websocket::close_code::normal,
          [self](boost::system::error_code ec) {
            if (auto s = self.lock()) {
              s->fail_async2(ec);
            } else {
              LOG(ERROR) << "WsConnection is already destroyed";
            }
          });
    }
  } else {
    fail_async2(ec);
  }
}

void WsConnection::Send(const std::string& msg) {
  if (!running_) return;

  if (sending_) {
    send_stack_.push(msg);
    return;
  }
  std::weak_ptr<WsConnection> self(shared_from_this());
  if (use_ssl_) {
    ws_ssl_->async_write(
        boost::asio::buffer(msg),
        [self, msg](const boost::beast::error_code& ec, size_t bytes_transferred) {
          if (auto s = self.lock()) {
            s->SendImpl(ec, msg);
          } else {
            LOG(ERROR) << "boost::asio handler object (WsConnection) does not exist";
          }
        });
  } else {
    ws_nossl_->async_write(
        boost::asio::buffer(msg),
        [self, msg](const boost::beast::error_code& ec, size_t bytes_transferred) {
          if (auto s = self.lock()) {
            s->SendImpl(ec, msg);
          } else {
            LOG(ERROR) << "boost::asio handler object (WsConnection) does not exist";
          }
        });
  }
  if (msg.find("pong") == std::string::npos) {
    num_subscription_sent_++;
  }
  sending_ = true;
}

void WsConnection::SendImpl(boost::beast::error_code ec, const std::string& last_msg) {
  if (ec) {
    fail_async1(ec, "error sending msg: " + last_msg);
  }
  if (!send_stack_.empty()) {
    auto msg = send_stack_.top();
    send_stack_.pop();
    std::weak_ptr<WsConnection> self(shared_from_this());
    if (use_ssl_) {
      ws_ssl_->async_write(
          boost::asio::buffer(msg),
          [self, msg](const boost::beast::error_code& ec, size_t bytes_transferred) {
            if (auto s = self.lock()) {
              s->SendImpl(ec, msg);
            } else {
              LOG(ERROR) << "WsConnection is already destroyed";
            }
          });
    } else {
      ws_nossl_->async_write(
          boost::asio::buffer(msg),
          [self, msg](const boost::beast::error_code& ec, size_t bytes_transferred) {
            if (auto s = self.lock()) {
              s->SendImpl(ec, msg);
            } else {
              LOG(ERROR) << "WsConnection is already destroyed";
            }
          });
    }
    if (msg.find("pong") == std::string::npos) {
      num_subscription_sent_++;
    }
  } else {
    sending_ = false;
  }
}

void WsConnection::SchedulePingPong() {
    // Schedule sending periodic ping if needed.
  if (!config_.mea_config->periodic_ping_msg.empty()) {
    DLOG(INFO) << config_.topic << "_" << config_.url.name << ", ping";
    scheduler_->Start();
    std::weak_ptr<WsConnection> self(shared_from_this());
    scheduler_->AddPeriodicTask(
        [self] {
          if (auto s = self.lock()) {
            s->Send(s->config_.mea_config->periodic_ping_msg);
          } else {
            LOG(ERROR) << "WsConnection handler is already destroyed";
          }
        },
        config_.mea_config->ping_interval_ns,
        "customized ping");
  }
}

void WsConnection::CleanUp() {
  running_ = false;    // stop receive data
  scheduler_->Stop();  // stop ping and unsubscribe

  // until close is done, the coroutine is alive
  if (!ws_closed_) {
    ws_closed_ = true;
    boost::system::error_code ec;
    boost::asio::ip::tcp::socket* socket = nullptr;
    if (use_ssl_) {
      socket = &boost::beast::get_lowest_layer<WebsocketSsl>(*ws_ssl_).socket();
    } else {
      socket = &boost::beast::get_lowest_layer<WebsocketNoSsl>(*ws_nossl_).socket();
    }
    socket->close(ec);
  }
}

void WsConnection::PostCleanUp() {
  std::weak_ptr<WsConnection> self(shared_from_this());
  boost::asio::post(
      strand_,
      [self] {
        if (auto s = self.lock()) {
          s->CleanUp();
        }
      });
}

}  // namespace impl
}  // namespace coin2::exchange::base::feed
